package com.gzyj.flink.driver.sink;

import com.alibaba.fastjson.support.spring.FastJsonRedisSerializer;
import com.gzyj.flink.driver.Driverlogininfo;
import com.gzyj.flink.driver.config.RedisKeyPrefixConfig;
import com.gzyj.flink.driver.config.RedisSessionFactory;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.sql.Connection;
import java.sql.Date;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.concurrent.TimeUnit;

/**
 * @author lhj
 */
public class DriverInfoSink extends RichSinkFunction<Driverlogininfo> {

    private Connection connection;
    private PreparedStatement ps;

    private LettuceConnectionFactory redisSessionFactory;
    private RedisTemplate<String,Object> getRedisTemplate(LettuceConnectionFactory redisSessionFactory){
        RedisTemplate<String,Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisSessionFactory);
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new FastJsonRedisSerializer<>(Object.class));
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(new FastJsonRedisSerializer<>(Object.class));
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }


    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        try {
            Class.forName("com.mysql.cj.jdbc.Driver");
            ExecutionConfig.GlobalJobParameters globalJobParam=getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
            Configuration globconf=(Configuration)globalJobParam;
            this.connection = DriverManager.getConnection(globconf.getString("mysql-url",null),
                    globconf.getString("mysql-user","root"),
                    globconf.getString("mysql-password","123456"));
            this.ps = this.connection.prepareStatement("insert into driverlogininfo values (?,?,?,?,?,?)");
            redisSessionFactory= RedisSessionFactory.getRedisFactory();
        } catch (Exception e) {
            System.out.println("-----------mysql get connection has exception , msg = " + e.getMessage());
        }



    }


    @Override
    public void invoke(Driverlogininfo value, Context context) throws Exception {
        ps.setInt(1,value.getCheckObject());
        ps.setInt(2,value.getLoginResults());
        ps.setString(3,value.getDriverName());
        ps.setString(4,value.getDriverNo());
        ps.setDate(5,value.getLoginTime()!=null? new Date(value.getLoginTime().getTime()):new Date(System.currentTimeMillis()));
        ps.setString(6,value.getVehId());
        ps.execute();

        RedisTemplate<String,Object> clusterRedisTemplate=getRedisTemplate(this.redisSessionFactory);
        if(value.getLoginResults()==0x01){
            clusterRedisTemplate.opsForValue().set(RedisKeyPrefixConfig.RedisKey_DriverSignOn+value.getVehId(),value);
        }

    }

    @Override
    public void close() throws Exception {
        super.close();
        if (ps != null) ps.close();
        if (connection != null) connection.close();


    }
}
